package com.ds.infrastructure.hyperspace.container.flow.operation;

import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.NettyEndpointNodeStream;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetProtocolAdapter;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.listener.HyperspaceEventListener;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import lombok.extern.slf4j.Slf4j;

import static com.ds.infrastructure.hyperspace.common.StateEnum.TARGET_SERVER_ERROR;

/**
 * @author  Weishaoying
 */
@Slf4j
public class HyperspaceTargetFlowOperation extends AbstractFlowOperation<HyperspaceDuplexFlowContext>
        implements TargetFlowOperation<HyperspaceDuplexFlowContext> {

    private final NettyContainer nettyContainer;

    private final HyperspaceEventListener<HyperspaceDuplexFlowContext> eventListener;

    private final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

    public HyperspaceTargetFlowOperation(NettyContainer nettyContainer,
            HyperspaceEventListener<HyperspaceDuplexFlowContext> eventListener,
            EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation) {
        this.nettyContainer = nettyContainer;
        this.eventListener = eventListener;
        this.entryFlowOperation = entryFlowOperation;
    }

    @Override
    public void handleInput(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        eventListener.fireTargetInput(duplexFlowContext, data);
        duplexFlowContext.endGatewayDuration();
    }

    @Override
    public void handleInputComplete(HyperspaceDuplexFlowContext duplexFlowContext) {
        eventListener.fireTargetInputComplete(duplexFlowContext);
        duplexFlowContext.endGatewayDuration();
    }

    @Override
    public void handleOutput(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        try {
            duplexFlowContext.getFlowHandlerPipeline().handleOutput(duplexFlowContext, data);
        } finally {
            duplexFlowContext.addPayloadSize(calculateMessageSize(data));
            duplexFlowContext.endGatewayDuration();
            eventListener.fireTargetOutput(duplexFlowContext, data);
            if (duplexFlowContext.isClosed()) {
                duplexFlowContext.getEntryProtocolAdapter().releaseData(data);
            }
        }
    }

    @Override
    public void handleOutputComplete(HyperspaceDuplexFlowContext duplexFlowContext) {
        eventListener.fireTargetOutputComplete(duplexFlowContext);
        entryFlowOperation.handleOutputComplete(duplexFlowContext);
    }

    @Override
    public void handleChannelActive(TargetEndpointNode targetEndpointNode,
            NettyEndpointNodeStream stream, TargetProtocolAdapter targetProtocolAdapter) {
        eventListener.fireTargetChannelActive(targetEndpointNode);
        stream.onConnectSuccess(targetProtocolAdapter);
    }

    @Override
    public void handleChannelInactive(TargetEndpointNode targetEndpointNode,
            NettyEndpointNodeStream stream, TargetProtocolAdapter targetProtocolAdapter,
            HyperspaceDuplexFlowContext duplexFlowContext) {
        eventListener.fireTargetChannelInactive(targetEndpointNode);
        nettyContainer.decrementTargetConnectionSize(targetEndpointNode);
        stream.onTargetChannelClose();
        targetProtocolAdapter.clearSuspendData();
        if (duplexFlowContext == null) {
            return;
        }
        if (duplexFlowContext.getEndpointProtocol().isLongLive()) {
            if (duplexFlowContext.getEndpointProtocol() == EndpointProtocolEnum.HTTP2
                    || duplexFlowContext.getEndpointProtocol() == EndpointProtocolEnum.HTTP2C) {
                duplexFlowContext.getEntryProtocolAdapter().releaseDuplexFlowContext();
                duplexFlowContext.getReferTargetEndpointStreamMap().remove(duplexFlowContext.getZoneId());
                //测试：http2连接抖动问题,目前后端连接关闭后，不再关闭前端连接。
                /*if (duplexFlowContext.getReferTargetEndpointStreamMap().size() == 0) {
                    duplexFlowContext.getEntryProtocolAdapter().closeEntryChannel();
                }*/
            } else {
                duplexFlowContext.getEntryProtocolAdapter().closeEntryChannel();
            }
        } else if (!duplexFlowContext.isClosed() && !duplexFlowContext.isOutputComplete() &&
                !duplexFlowContext.isPending()) {
            duplexFlowContext.setTargetResponseState(
                    null, TARGET_SERVER_ERROR.getHttpCode());
            interruptAndCloseContext(duplexFlowContext, TARGET_SERVER_ERROR,
                    String.format("Target connection closed but response incorrectly! target node(%s),remote address:%s",
                            targetEndpointNode.getKey(), stream.getRemoteAddress()));
        }
    }

    @Override
    public void handleExceptionCaught(String targetEndpointName, Throwable cause) {
        eventListener.fireTargetExceptionCaught(targetEndpointName, cause);
    }

    @Override
    public void handlePendingRequestIncrement(String targetEndpointName, String zone) {
        eventListener.firePendingRequestIncrement(targetEndpointName);
        nettyContainer.incrementTargetEndpointPendingSize(targetEndpointName, zone);
    }

    @Override
    public void handlePendingRequestDecrement(String targetEndpointName, String zone) {
        eventListener.firePendingRequestDecrement(targetEndpointName);
        nettyContainer.decrementTargetEndpointPendingSize(targetEndpointName, zone);
    }

    @Override
    public void handleTargetNodeStatusChange(String targetEndpointName, String nodeInfo, int status) {
        eventListener.fireTargetNodeStatusChange(targetEndpointName, nodeInfo, status);
    }

}
